package com.shujia.stream

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.client.{HConnectionManager, Put}
import org.apache.hadoop.hbase.util.Bytes
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Demo9ToHbase {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("stream")
      .setMaster("local[2]") //指定两个线程
      .set("spark.sql.shuffle.partitions", "4")
    val sc = new SparkContext(conf)

    //创建spark streaming上下文对象   指定间隔时间
    val ssc = new StreamingContext(sc, Durations.seconds(5))

    ssc.checkpoint("data/checkpoint")

    val topics = Map("topic4" -> 2) //2  两个线程读数据

    //连接kafka创建DS  key 偏移量   value  数据
    val kakfaDS = KafkaUtils.createStream(
      ssc,
      "node1:2181,node2:2181,node3:2181",
      "123123",
      topics
    ).map(_._2)
      .filter(line => {
        val city = line.split(",")(2)
        !city.equals("\\N")
      })

    /**
      * 实时统计城市客流量，将统计结果写入hbase
      *
      * 创建表
      * create 'city_count','info'
      *
      */


    kakfaDS.map(line => {
      val cityId = line.split(",")(2)
      (cityId, 1)
    })
      .updateStateByKey((seq: Seq[Int], opt: Option[Int]) => Option(seq.sum + opt.getOrElse(0)))
      .foreachRDD(rdd => {
        rdd.foreachPartition(iter => {
          //创建连接  每一个分区一个连接
          //创建hbase连接  将数据写入hbase
          val conf: Configuration = new Configuration()
          conf.set("hbase.zookeeper.quorum", "node1:2181,node2:2181,node3:2181")
          val connection = HConnectionManager.createConnection(conf)

          //获取表连接
          val table = connection.getTable("city_count")

          iter.foreach(kv => {
            val city = kv._1
            val count = kv._2

            //构建put对象
            val put = new Put(city.getBytes())
            put.add("info".getBytes(), "count".getBytes(), Bytes.toBytes(count))

            //插入数据
            table.put(put)

          })

          //关闭连接
          connection.close()


        })

      })


    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }
}
